package org.jerrymouse.jsa4j.db.kv.cassandra;

import java.util.HashMap;
import java.util.Map;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Cassandra.Client;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.jerrymouse.jsa4j.db.kv.DB;
import org.jerrymouse.jsa4j.db.kv.DBManager;

public class CassandraDBManager implements DBManager {
	private Map<String, String> properties = new HashMap<String, String>();
	@SuppressWarnings("unused")
	private static Log log = LogFactory.getLog(CassandraDBManager.class);

	@Override
	public DB getDB() {
		CassandraDB cassandraDB = new CassandraDB();
		cassandraDB.setClient(getClient());
		return cassandraDB;
	}

	@Override
	public void setProperties(Map<String, String> properties) {
		this.properties = properties;
	}

	public Map<String, String> getProperties() {
		if (properties == null) {
			properties = new HashMap<String, String>();
		}
		return properties;
	}

	private String getHost() {
		return getProperty("host", "localhost");
	}

	private String getProperty(String name, String defaultValue) {
		if (getProperties().get(name) == null) {
			getProperties().put(name, defaultValue);
		}
		return getProperties().get(name);
	}

	private int getPort() {
		return Integer.parseInt(getProperty("port", "9160"));

	}

	public Client getClient() {
		Client client = null;

		try {
			TTransport tr = new TSocket(getHost(), getPort());
			TProtocol proto = new TBinaryProtocol(tr);
			client = new Cassandra.Client(proto);
			tr.open();
		} catch (TTransportException e) {
			e.printStackTrace();
		}
		return client;

	}

}
